package org.example.producerconfirm;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: anlu
 * @Date: 2021/06/22/14:31
 * @Description: 存放到延迟队列中，对业务数据进行包装
 */
public class ProducerConfirmAsync {
    public final static String EXCHANGE_NAME = "Exchange_producer_async_confirm";
    public final static String QUEUE_NAME = "ProducerConfirmAsync_Queue";
    public static void main(String[] args)  throws IOException, TimeoutException, InterruptedException{
        // 创建和连接rabbitmq
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        // 创建一个连接
        Connection connection = factory.newConnection();
        // 创建一个信道
        Channel channel = connection.createChannel();
        // 指定转发的路由器
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        // 启用发送者确认模式
        channel.confirmSelect();
        /**
         * 消费者确认消息的回调方法
         */
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple)
                    throws IOException {
                System.out.println("=========handleAck deliveryTag:"+deliveryTag
                        +",multiple:"+multiple);

            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple)
                    throws IOException {

            }
        });
        /**
         * mandatory：消息发送失败的回调方法
         */
        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText,
                                     String exchange, String routingKey,
                                     AMQP.BasicProperties properties,
                                     byte[] body)
                    throws IOException {
                String message = new String(body);
                System.out.println("RabbitMq路由失败:路由键是："+routingKey+"."+message);
            }
        });

        String[] severities={"error","warning"};
        for(int i=0;i<100;i++){
            String severity = severities[i%2];
            // 发送的消息
            String message = "Hello World_"+(i+1)+("_"+System.currentTimeMillis());
            channel.basicPublish(EXCHANGE_NAME, severity, true,
                    MessageProperties.PERSISTENT_BASIC, message.getBytes());
            System.out.println("----------------------------------------------------");
            System.out.println(" Sent Message "+i+" : [路由键:" + severity +"]:'"+ message + "'");
        }
    }


}
